package mono;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.testng.annotations.Test;
import reactor.core.publisher.Mono;

public class Mono1 {


    @Test(invocationCount = 1)
    private void method1() {

        AtomicInteger a = new AtomicInteger();
        AtomicReference<Object> b = new AtomicReference<>();
        for (int i = 0; i < 5; i++) {

            Mono<Integer> stringMono = Mono.fromCompletionStage(new Mono2().createNum());

            AtomicInteger flag = new AtomicInteger();
            stringMono.doOnNext(res -> {
                System.out.println("execute mono");
                if (res > 4) {
                    flag.set(1);
                    a.set(res);
                }
            }).subscribe();

            System.out.println("execute flag");
            if (flag.get() == 1) {
                System.out.println(i);
                break;
            }
        }

        System.out.println("-------------------");
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 5; i++) {

            Mono<Integer> stringMono = Mono.fromCompletionStage(new Mono2().createNum());

            AtomicInteger flag = new AtomicInteger();
            stringMono.doOnNext(res -> {
                System.out.println("mono " + Thread.currentThread().getName());
                if (res > 4) {
                    flag.set(1);
                }
            })
            .subscribe();

//            stringMono.map(res -> {
//                System.out.println("execute mono");
//                if (res > 4) {
//                    flag.set(1);
//                    return 1;
//                }
//                return 0;
//            });

            System.out.println("execute flag " + Thread.currentThread().getName());
            if (flag.get() == 1) {
                System.out.println(i);
                break;
            }
        }

        Thread.sleep(1000);
    }
}
